package my.thesis.spouts;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import org.json.JSONArray;
import org.json.JSONObject;

import twitter4j.FilterQuery;
import twitter4j.GeoLocation;
import twitter4j.HashtagEntity;
import twitter4j.Place;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.User;
import twitter4j.UserMentionEntity;
import twitter4j.conf.ConfigurationBuilder;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class TweetReader2 extends BaseRichSpout
{private LinkedBlockingQueue<Status> queue = null;
private TwitterStream twitterStream;
private int queueDepth;
private SpoutOutputCollector collector;
private int counter=0;

//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********

File file;
FileWriter fw;
BufferedWriter bw;

//************* SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT ********

public void nextTuple() 
{
	
	//retrieves and removes head of queue 
	Status ret = queue.poll();
	
	if(ret == null)
	{
		try 
		{ 
			Thread.sleep(50); 
		}
		catch (InterruptedException e) {}
	} 
	else 
	{	        
		collector.emit(new Values(ret));
	}
	
}

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) 
{
	queueDepth=2000;
	this.collector = collector;
	queue = new LinkedBlockingQueue<Status>(queueDepth);
	
	//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********
	
	File file = new File("MY_TWEETS_KEYWORD_TEST.json");

	if(!file.exists())
	{
		try {
			file.createNewFile();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	try 
	{
		fw = new FileWriter(file.getAbsoluteFile());
	} 
	catch (IOException e1) 
	{
		
		e1.printStackTrace();
	}
	
	bw = new BufferedWriter(fw);
	
	//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********
	
	 StatusListener listener = new StatusListener()
	 {

		public void onException(Exception arg0) {
			// TODO Auto-generated method stub
			
		}

		public void onDeletionNotice(StatusDeletionNotice arg0) {
			// TODO Auto-generated method stub
			
		}

		public void onScrubGeo(long arg0, long arg1) {
			// TODO Auto-generated method stub
			
		}

		public void onStallWarning(StallWarning arg0) {
			// TODO Auto-generated method stub
			
		}

		public void onStatus(Status status) 
		{
			if(queue.size() < queueDepth)
    		{
    			//System.out.println("TWEET Received: " + status.getText());
    			
    			//inserts at tail of queue
    			queue.offer(status);
    			    				
    				User user =status.getUser();
    				
    				String username = user.getName();
    						
    				int retweets=status.getRetweetCount();
    				
    				int followers=user.getFollowersCount();
    				
    				UserMentionEntity[] mentions=status.getUserMentionEntities();
    			
    				if(mentions.length > 0)
    				{
    					for(int i=0; i < mentions.length ; i++)
    					{
    						UserMentionEntity h = mentions[i];				
    						
    					}
    				}
 				
    				Date created = status.getCreatedAt();
    				long millis=created.getTime();
    				
    				String createdat=created.toString();
    				
    				String lang = status.getLang();
    				
    				String text = status.getText();
    				
    				GeoLocation geo = status.getGeoLocation();
    				
    				double lon = 0;
    				double lat = 0;
    				
    				if(geo!=null)
    				{
    					lat=geo.getLatitude();
    					lon=geo.getLongitude();
    				}
    				else
    				{
    					lat=Math.random()*-90 + Math.random()*90;
    					lon=Math.random()*-180+Math.random()*180;    					
    				}
    							
    				Place pla = status.getPlace();
    				
    				String country = "";
    				
    				if(pla!=null)
    				{
    					if(pla.getPlaceType()!=null)
    					{				
    						country=pla.getCountry();
    					}
    				
    				}
    				
	    			HashtagEntity[] hashtags = status.getHashtagEntities();	
	    				
	    			if(hashtags.length > 0)
	    			{
	    				for(int i=0; i < hashtags.length ; i++)
	    				{
	    					HashtagEntity h = hashtags[i];				
	    						
	    				}
	    			}
	    	
	    				//******************** MY FORM **************************	 
	    				
	    				JSONArray hashtagsArray = new JSONArray();
	    				
	    				for(HashtagEntity hash : hashtags)
	    				{
	    					JSONObject hashObj= new JSONObject();
	    					hashObj.put("text", hash.getText());
	    					
	    					hashtagsArray.put(hashObj);
	    				}
	    				
	    				JSONArray mentionsArray = new JSONArray();
	    				
	    				for(UserMentionEntity mention : mentions)
	    				{
	    					JSONObject mentionObj= new JSONObject();
	    					mentionObj.put("text", mention.getName());
	    					
	    					mentionsArray.put(mentionObj);
	    				}

	    				JSONObject tweetObject = new JSONObject();
	    						
	    				tweetObject.put("username",username);
	    				tweetObject.put("retweets",retweets);
	    				tweetObject.put("followers",followers);
	    				
	    				tweetObject.put("lang", lang);
	    				tweetObject.put("createdat",createdat);
	    				tweetObject.put("millis", millis);
	    				tweetObject.put("text",text);
	    				tweetObject.put("lat", lat);
	    				tweetObject.put("lon",lon);
	    				tweetObject.put("country", country);
	    				
	    				tweetObject.put("hashtags", hashtagsArray);
	    				tweetObject.put("mentions",	 mentionsArray);
	    				
	    				String mytweet = tweetObject.toString();
	    				System.out.println("Counter: " + counter);
	    				System.out.println(mytweet);
	    				counter++;
		    	        try 
		    	        {
							bw.write(mytweet);
							bw.newLine();
						} 
		    	        catch (IOException e) 
		    	        {
							// TODO Auto-generated catch block
							e.printStackTrace();
		    	        }
    				
	    	        
    				//******************** MY FORM *******************************	
    				
    				//******************** TWITTER FORM **************************	    

/*	    				String rawJSON = TwitterObjectFactory.getRawJSON(status);
	    	        System.out.println(rawJSON);
	    	        counter++;
	    	        System.out.println("Counter: " + counter);
	    	        try {
						bw.write(rawJSON);
						bw.newLine();
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}*/
	    	      //******************** TWITTER FORM ****************************	
    			//}
    		} 
			else 
    		{
    			System.out.println("Queue is now full, the following message is dropped: " + status.getText());
    		}			
		}

		public void onTrackLimitationNotice(int arg0) {
			// TODO Auto-generated method stub
			
		}
	 
	 };
	 		 
	    ConfigurationBuilder cb = new ConfigurationBuilder();

	    cb.setDebugEnabled(true)
	      .setOAuthConsumerKey("6eHKCDId2GB72YsMrvgYeWVtU")
	      .setOAuthConsumerSecret("VDk8xgW3TElb5ErTLJcdImu3w5pVMs6Frm23SOeB8wYlPiI61a")
	      .setOAuthAccessToken("2462025612-2aOGrd3ZFZ1tv3Xfn4iCB3DggRoPPUFtRLxmyEL")
	      .setOAuthAccessTokenSecret("8pU9R2wNUmf7ghj1kL7cEqdnOBYKpA9AHxRQuaGcQaCCJ");
	    
	    cb.setJSONStoreEnabled(true);
	    cb.setIncludeEntitiesEnabled(true);
	 
	    twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
	    twitterStream.addListener(listener);
	    
	    FilterQuery filter = new FilterQuery();
	    String[] keywordsArray = { "big data"};
	    filter.track(keywordsArray);
	
	    twitterStream.filter(filter);

	    
}

public void declareOutputFields(OutputFieldsDeclarer declarer) 
{
	declarer.declare(new Fields("tweet"));
	
}

}
